package com.daona.job.kafka;

import com.daona.job.service.PdfJobFacade;
import com.daona.model.constants.Constants;
import com.daona.model.message.PdfMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
@Slf4j
@RequiredArgsConstructor
public class PdfMessageListener {

    private final PdfJobFacade jobFacade;

    @KafkaListener(groupId=Constants.GROUP_ID, topics = Constants.TOPIC_PDF)
    public void listenPdfMessage (@Payload ConsumerRecord<String,PdfMessage> record, Acknowledgment acknowledgment) {
        try {
            String key = record.key();
            PdfMessage body = record.value();
            log.info("\n=====\ntopic:{}，key{}，body:{}\n=====\n",Constants.TOPIC_PDF,key,body);
            jobFacade.PdfJobProcess(body);
        } catch (Exception e){
            log.error(e.getMessage());
        } finally {
            acknowledgment.acknowledge();
            log.info("acknowledge successfully");
        }
    }
}
